package com.atguigu.gmall.app.dwd.log;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.util.DateFormatUtils;
import com.atguigu.gmall.util.MyKafkaUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

/**
 * 流量域 — 独立访客明细表
 * 数据流: web/app -> Nginx -> 日志服务器(.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> Kafka(DWD)
 * 程  序: Mock(lg.sh) -> Flume(f1) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> DwdTrafficUniqueVisitorDetail -> Kafka(ZK)
 *
 * @author : ranzlupup
 * @since : 2023/6/2 15:08
 */
public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {
        //! 1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境中设置为Kafka主题的分区数

        //1.1 开启CheckPoint
        //env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));

        //1.2 设置状态后端
        //env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //! 2.读取Kafka 页面日志主题创建流
        String topicName = "FLINK_DWD_PAGE_LOG";
        String groupId = "FLINK_DWD_PAGE_LOG_UNIQUE_VISITOR_DETAIL";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = MyKafkaUtils.getFlinkKafkaConsumer(topicName, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(flinkKafkaConsumer);

        //! 3.过滤掉上一跳页面不为null的数据并将每行数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjWithNullDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                    if (lastPageId == null) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    System.out.println("发现脏数据: " + value);
                }
            }
        });

        //! 4.按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithNullDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //! 5.使用状态编程实现按照Mid的去重
        SingleOutputStreamOperator<JSONObject> filterUVDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> lastVisitState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<String>("last-visit-state", String.class);

                //? 设置状态生命周期
                // 1天过期，创建和写入的时候更新过期时间
                // StateTtlConfig ttlConfig = StateTtlConfig
                //         .newBuilder(Time.days(1))
                //         .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                //         .build();
                // stateDescriptor.enableTimeToLive(ttlConfig);

                StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();
                stateDescriptor.enableTimeToLive(ttlConfig);

                lastVisitState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
                // 获取状态数据 & 当前数据中的时间戳 并 转换为日期
                String lastDate = lastVisitState.value();
                Long ts = value.getLong("ts");
                String curDate = DateFormatUtils.toDate(ts);

                System.out.println("状态更新前：" + lastDate);

                if (lastDate == null || !lastDate.equals(curDate)) {
                    // 如果状态为null，或者 状态不为null同时当前日期不等于上次浏览的日期（不是同一天）
                    // 更新状态，收集数据 (返回true)
                    lastVisitState.update(curDate);

                    System.out.println("状态更新后：" + lastVisitState.value());

                    return true;
                } else {
                    return false;
                }
            }
        });

        //! 6.将数据写到Kafka
        String targetTopic = "FLINK_DWD_TRAFFIC_UNIQUE_VISITOR_DETAIL";
        filterUVDS.print(">>>>>>>>");
        filterUVDS
                .map(JSONAware::toJSONString)
                .addSink(MyKafkaUtils.getFlinkKafkaProducer(targetTopic));

        //! 7.启动任务
        env.execute("DwdTrafficUniqueVisitorDetail => ");
    }

}
